package distributedbank;
import java.net.Socket;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.io.BufferedReader;
import java.io.DataInputStream;
import java.io.FileInputStream;
import java.io.InputStreamReader;
import java.io.ObjectOutputStream;
import java.io.Serializable;
import java.io.IOException;

/**
 * The wrapper parses the topology file and takes charge of serializing and
 * sending message objects. The Wrapper will refuse to send messages to
 * dis-allowed JVMs.
 * 
 *
 * Note that we require all JVM names in the topology file to be of the form
 * "JVM#" where "#" is a digit.
 */
public class Wrapper {
    /**
     * each JVM will listen on port = (basePort + JVM Number)
     */
    public static final int basePort = 20000;

    /**
     * Map JVM number to a list of JVM Numbers it is able to reach
     */
    private static Map<Integer, List<Integer>> topologyMapping;

    /**
     * static constructor
     */
    static {
        topologyMapping = new HashMap<Integer, List<Integer>>();
        readTopology();
    }

    /**
     * Read topology_file.txt
     */
    private static void readTopology() {
        try {
            FileInputStream fstream = new FileInputStream("topology_file.txt");
            DataInputStream in = new DataInputStream(fstream);
            BufferedReader br = new BufferedReader(new InputStreamReader(in));
            String strLine;
            while ((strLine = br.readLine()) != null) {
                updateTopologyMap(Integer.parseInt(String.valueOf(strLine.charAt(3))),
                                  Integer.parseInt(String.valueOf(strLine.charAt(8))));
            }
            in.close();
        } catch (Exception e) {
            System.err.println("Error: " + e.getMessage());
        }
    }

    /**
     * Insert one mapping information into topologyMapping
     */
    private static void updateTopologyMap(int src, int sink) {
        if (topologyMapping.containsKey(src)) {
            topologyMapping.get(src).add(sink);
        } else {
            List<Integer> list = new ArrayList<Integer>();
            list.add(sink);
            topologyMapping.put(src, list);
        }
    }

    /**
     * Attempts to send a Message to the JVM with the given id. Returns true if
     * the it is permitted to send a message to the given JVM. Otherwise
     * returns false.
     */
    public boolean sendMessage(Serializable message, int srcJvmId, int dstJvmId) {
        //cannot reach directly
        if (!whoNeighbors(srcJvmId).contains(dstJvmId)) {
            return false;
        } else {
            try {
                Socket socket = new Socket("localhost", dstJvmId + basePort);
                ObjectOutputStream out =
                        new ObjectOutputStream(socket.getOutputStream());
                try {
                    out.writeObject(message);
                } finally {
                    out.close();
                    socket.close();
                }
            } catch (IOException e) {
                System.out.println(e.getMessage());
            }
            return true;
        }
    }

    /**
     * Look up neighbors
     * @param JVM Number
     * @return the list of JVM numbers that the given JVMNumber is able to reach directly
     */
    public static List<Integer> whoNeighbors(int JVMNumber) {
        List<Integer> neighbors = topologyMapping.get(JVMNumber);
        return neighbors == null ? new ArrayList<Integer>() : neighbors;
    }

}
